package com.ld.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author ld
 *
 * 生产者客户端示例
 */
public class KafkaProducerDemo {

    //kafka服务器地址
    private static String servers = "192.168.150.101:9092,192.168.150.102:9092,192.168.150.103:9092";
    //客户端id
    private static String clientId = "product.client.demo";
    //主题名称
    private static String topic = "first";


    /**
     * 生产者客户端配置
     * 这里只配置了几个必要的参数，其他参数后面会一一介绍。
     * @return
     */
    private static Properties config(){
        Properties properties = new Properties();

        /*************************优化前*****************************/
//        //kafka服务器地址
//        properties.put("bootstrap.servers",servers);
//        //key的序列化器类路径
//        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
//        //value的序列化器类路径
//        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
//        //客户端id
//        properties.put("client.id",clientId);

        /*************************优化后*****************************/
        /**
         * 为了防止开发人员将参数名写错，所以客户端提供了一个 ProducerConfig 类，这个类中定义了所有要配置的参数名。
         * 所以我们可以使用这个类中的定义参数名。
         */
        //kafka服务器地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
        //key的序列化器类路径,序列化器可以直接获取类路径
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //value的序列化器类路径
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //客户端id
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,clientId);

        //返回配置对象
        return properties;
    }


    public static void main(String[] args) {
        //获取客户端配置
        Properties properties = config();
        //创建KafkaProducer对象,传入客户端配置信息
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //创建ProducerRecord对象,传入主题名称，消息
        ProducerRecord<String, String> record = new ProducerRecord<>(topic,"Hello World");

        //发送ProducerRecord对象
        try{
            //普通发送
            producer.send(record);

            //同步发送
//            RecordMetadata recordMetadata = producer.send(record).get();
//            //打印发送结果
//            System.out.println("topic:" + recordMetadata.topic() + ",partition:" + recordMetadata.partition() + ", offset" + recordMetadata.offset());

            //异步发送
//            producer.send(record, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                    //判断如果recordMetadata 不为空，说明发送成功，否则发送失败，打印异常。
//                    if(recordMetadata != null){
//                        System.out.println("topic:" + recordMetadata.topic() + ",partition:" + recordMetadata.partition() + ", offset" + recordMetadata.offset());
//                    }else{
//                        e.printStackTrace();
//                    }
//                }
//            });
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //关闭KafkaProducer对象
            producer.close();
        }
    }
}
